package com.yujiahao.bigdata.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_TestHotTop10_2 {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接

    val conf: SparkConf = new SparkConf().setAppName("TestTop10").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑


    //TODO 需求一的问题：优化方案三
    // 在读取三个数量的时候，尽量减少shuffle，所以就会提高一定的性能
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")
    //这里用扁平化，因为下单可能会有多个
    rdd.flatMap(
      data => {
        //TODO 注意：flatMap会返回一个迭代器，因此要用集合接收一下，接着使用reduceByKey进行聚合
        val datas: Array[String] = data.split("_")
        //如果第六个数据不等于-1的话那么就是点击的数据
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
          //如果第8个数据不是null的话那么就是下单数量
        } else if (datas(8) != "null") {
          //这里注意第八个可能会有好多个要用逗号切分开然后再使用map转换格式
          val ids: Array[String] = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
          //如果第10个数据不是null的话那么就是支付数量
        } else if (datas(10) != "null") {
          //这里注意第十个可能会有好多个要用逗号切分开然后再使用map转换格式
          val ids: Array[String] = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
          //否则返回一个空集合
        } else {
          Nil
        }
      }
    ).reduceByKey( //这里一样的结果两两聚合
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    ).sortBy(_._2, false).take(10).foreach(println)

    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

}
